package thread

import (
	"fmt"
	"strings"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/logger"

	"ssdevice/common/constant"
	"ssdevice/internal/device/models"
)

type DevSensorDataThread struct {
}

func (DevSensorDataThread) Run(threadName string) {
	logger.Infof("启动线程 - %s.", threadName)

	go func() {
		dataList := make([]models.SysDevSensor, 0)
		for {
			time.Sleep(1 * time.Second)
			for {
				s := constant.SensorQueue.Dequeue()
				if s == nil {
					break
				}
				dataList = append(dataList, s.(models.SysDevSensor))
				if len(dataList) > 1000 {
					// 批量处理数据
					dealDevSensorData(dataList)
					// 清空(减少扩容)
					dataList = dataList[0:0]
				}
			}
			if len(dataList) > 0 {
				// 批量处理数据
				dealDevSensorData(dataList)
				// 清空(减少扩容)
				dataList = dataList[0:0]
			}
		}
	}()
}

var scan = map[string]interface{}{}

func dealDevSensorData(dataList []models.SysDevSensor) {
	t1 := time.Now()
	sql := "UPDATE sys_dev_sensor SET data = tmp.data FROM (VALUES %s) as tmp (sn, data) where sys_dev_sensor.dev_sn = tmp.sn"
	valueSql := make([]string, 0)
	for _, s := range dataList {
		valueSql = append(valueSql, "('"+s.DevSn+"', '"+string(s.Data)+"'::jsonb)")
	}
	sql = fmt.Sprintf(sql, strings.Join(valueSql, ","))
	if err := env.Db.Raw(sql).Scan(&scan).Error; err != nil {
		logger.Error(err)
	}
	logger.Infof("实时sensor批量更新%d. 耗时%v", len(dataList), time.Since(t1).Round(time.Millisecond))
}
